package com.doit.day01;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.record.TimestampType;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;

public class ConsumerDemo {
    public static void main(String[] args) {
        Properties props = new Properties();
        /**
         * 必配的
         */
        props.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"linux01:9092,linux02:9092,linux03:9092");
        //key和value的反序列化方式
        props.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        //消费者组
        props.setProperty(ConsumerConfig.GROUP_ID_CONFIG,"qwq");

        /**
         * 选配的
         */
        //是否允许自动创建topic
        props.setProperty("allow.auto.create.topics","true");
        //latest, earliest, none
        // 如果说，有组id，但是你已经指定了从什么地方开始消费，那么他就会从你指定的位置开始消费
        //如果说你没有指定位置开始消费，会去__consumer_offset 这个topic里面有之前的记录，那么就会失效，从上次消费到的地方开始消费
        // 如果没有，那么会按照下面设置的这个参数来执行
        props.setProperty("auto.offset.reset","earliest");

        //是否允许自动提交偏移量  __consumer_offset  全自动
        props.setProperty("enable.auto.commit","true");  //如果是true，有一定的可能会丢数据
        //如果设置成了false 那么他就不自动提交偏移量了  ==》手动提交   半自动  commitAsync  commitSync ==》 同样的是提交到  __consumer_offset

        //能否实现精准一次性消费呢？？？？



        //自动提交偏移量，每隔多长时间提交一次
//        props.setProperty("auto.commit.interval.ms","10000");

        //创建kafka的消费者对象
        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);

        //订阅主题  第一种方式
        consumer.subscribe(Arrays.asList("event-log"));

        //第二种方式 assign
/*        TopicPartition topicPartition = new TopicPartition("doit38", 0);
//        TopicPartition topicPartition1 = new TopicPartition("doit38", 1);
        //类似于订阅主题，只不过他分的更细，可以指定消费哪一个分区的数据
        consumer.assign(Arrays.asList(topicPartition));
        //指定从什么位置开始消费
        consumer.seek(topicPartition,10);*/




        while (true){
            //拉取broker中的数据
            ConsumerRecords<String, String> poll = consumer.poll(Duration.ofMillis(Integer.MAX_VALUE));

            //处理数据的业务逻辑
            for (ConsumerRecord<String, String> consumerRecord : poll) {
                String value = consumerRecord.value();
                String topic = consumerRecord.topic();
                int partition = consumerRecord.partition();
                long offset = consumerRecord.offset();
                /**
                 * 时间戳类型有两种：
                 *  1.数据写入到kafka中的时间（创建这条数据的时间）  默认
                 *  2.数据追加到日志里面的时间戳
                 */
                long timestamp = consumerRecord.timestamp();
                TimestampType timestampType = consumerRecord.timestampType();
                System.out.println("主题："+topic +",分区："+partition+",偏移量:"+offset+",值："+value+",时间戳："+timestamp+",时间戳类型:"+timestampType);
            }

            //同步提交偏移量
            consumer.commitSync();

        }



    }
}
